package com.mjf.transformation

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala._

/**
 * keyBy
 *  基于key的hash code重分区；
 *  同一个key只能在一个分区内处理，一个分区内可以有不同的key的数据
 *  keyBy之后的KeyedStream上的所有操作，针对的作用域都只是当前的key
 *
 * 聚合操作
 *  DataStream没有聚合操作，目前所有的聚合操作都是针对KeyedStream
 */
object KeyByDemo {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)

    val source: DataStream[String] = env.fromCollection(List(
      "hello world",
      "hello java",
      "hello scala",
      "hello hadoop",
      "spark  flink"
    ))

    source.print("source")

    // 方式一：通过传入函数使用 keyBy
    val result1: DataStream[(String, Int)] = source
      .flatMap(_.split("\\W+"))
      .map((_, 1))
      .keyBy(_._1) // 使用元组的第一个字段进行聚合
      .reduce((a, b) => (a._1, a._2 + b._2))


    // 方式二：通过传入自定义函数类使用 keyBy
    val result2: DataStream[(String, Int)] = source
      .flatMap(_.split("\\W+"))
      .map((_, 1))
      .keyBy(new MyKeySelector)
      .sum(1)

    result1.print("函数")
    result2.print("自定义函数类")

    env.execute(KeyByDemo.getClass.getName)

  }
}

class MyKeySelector extends KeySelector[(String, Int), String] {
  override def getKey(value: (String, Int)): String = value._1
}